package job;

import function.CarViewBeanMapFuction;
import net.bwie.realtime.jtp.common.utils.DorisUtil;
import net.bwie.realtime.jtp.common.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class KafkaDataToDoris {
    public static void main(String[] args) throws Exception {
        // env执行环境
        LocalStreamEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(3000L);
        // 消费Kafka
        DataStream<String> kafkaStream = KafkaUtil.consumerKafka(env, "new_car_clean_log");
        // 数据转换
        DataStream<String> resultStream = healde(kafkaStream);
        resultStream.print();
        // 存入Doris
        DorisUtil.saveToDoris(
                resultStream,
                "ner_car_report",
                "new_car_log_info"
        );
        // 触发执行
        env.execute("KafkaDataToDoris");
    }

    private static DataStream<String> healde(DataStream<String> stream) {
        // 转实体类
        SingleOutputStreamOperator<NewCarBean> beanStream = stream.map(new CarViewBeanMapFuction());
        // 设置水位线
        SingleOutputStreamOperator<NewCarBean> timeStream = beanStream.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<NewCarBean>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner(new SerializableTimestampAssigner<NewCarBean>() {
                            @Override
                            public long extractTimestamp(NewCarBean newCarBean, long l) {
                                return newCarBean.getTimestamp();
                            }
                        })
        );
        SingleOutputStreamOperator<String> processStream = beanStream.process(new ProcessFunction<NewCarBean, String>() {
            @Override
            public void processElement(NewCarBean newCarBean, Context context, Collector<String> collector) throws Exception {
                String eventType = newCarBean.getEventType();
                String province = newCarBean.getProvince();
                Double motorTemp = newCarBean.getMotorTemp();
                String vin = newCarBean.getVin();
                Double batteryTemp = newCarBean.getBatteryTemp();
                Long timestamp = newCarBean.getTimestamp();
                Double batteryPower = newCarBean.getBatteryPower();
                String chargingStatus = newCarBean.getChargingStatus();
                String startStatus = newCarBean.getStartStatus();
                Double speed = newCarBean.getSpeed();
                Double mileage = newCarBean.getMileage();
                String dateTime = newCarBean.getDateTime();
                String output = vin+","+eventType + "," + province + "," + motorTemp
                        + "," + batteryTemp + "," + timestamp + "," + batteryPower + "," + chargingStatus
                        + "," + startStatus + "," + speed + "," + mileage + "," + dateTime;
                collector.collect(output);
            }
        });
        return processStream;
    }
}
